/**
 * 串口转发json到MQTT，在finsh中运行
 * 
 * 
 * 
 * 
 * 
 */
#include <stdlib.h>
#include <string.h>
#include <stdint.h>

#include <rtthread.h>
#include "paho_mqtt.h"
#include "cJSON.h"

#define DEFAULT_UART_NAME "uart3"

#define MQTT_TEST_SERVER_URI "tcp://cyjason.top:1883"
#define MQTT_CLIENTID "serial2mqtt-rtthread"
//#define MQTT_USERNAME           "admin"
//#define MQTT_PASSWORD           "admin"
#define MQTT_SUBTOPIC "control"
#define MQTT_PUBTOPIC "test"
#define MQTT_WILLMSG "Goodbye!"
#define MQTT_TEST_QOS 1
#define MQTT_PUB_SUB_BUF_SIZE 1024

#define TEST_DATA_SIZE 256
#define PUB_CYCLE_TM 1000

static rt_thread_t pub_thread_tid = RT_NULL;
/*发布相关*/
static char *pub_data = RT_NULL;

/* define MQTT client context */
static MQTTClient client;

static int test_is_started = 0;

/* 进程间传递消息结构*/
struct threads_msg
{
    int state; //是否为新数据
    char uart_str[100];
};

struct threads_msg uart_msg;
static struct rt_semaphore mqtt_sem;

/* 用于接收消息的信号量 */
static struct rt_semaphore rx_sem;
static rt_device_t serial;

/* 接收数据回调函数 */
static rt_err_t uart_input(rt_device_t dev, rt_size_t size)
{
    /* 串口接收到数据后产生中断，调用此回调函数，然后发送接收信号量 */
    rt_sem_release(&rx_sem);
    return RT_EOK;
}

static void serial_thread_entry(void *parameter)
{
    char ch;
    char str[100] = {0}, buf[100] = {0};
    int nbit = 0;
    while (1)
    {
        /* 从串口读取1个字节的数据，没有读取到则等待接收信号量 */
        while (rt_device_read(serial, 0, str, 1) != 1)
        {
            /* 阻塞等待接收信号量，等到信号量后再次读取数据 */
            rt_sem_take(&rx_sem, RT_WAITING_FOREVER);
        }
        if (str[0] == '}')
        {
            buf[nbit] = '}';
            /* 读取到的数据通过串口输出 */
            rt_kprintf(buf);
            rt_kprintf("\r\n");

            /*写入结构体*/
            rt_strncpy(uart_msg.uart_str, buf, 99);
            uart_msg.state = 1;
            rt_sem_release(&mqtt_sem);

            nbit = 0;
        }
        else if (nbit == 0)
        {
            rt_strncpy(buf, "", 99);
            buf[nbit] = str[0];
            nbit++;
        }
        else
        {
            buf[nbit] = str[0];
            nbit++;
        }
    }
}

/*mqtt part*/

static void mqtt_sub_callback(MQTTClient *c, MessageData *msg_data)
{
    rt_kprintf("sub callback");
    return;
}

/**
 * This function create and config a mqtt client.
 *
 * @param void
 *
 * @return none
 */
static void mq_start(void)
{
    /* init condata param by using MQTTPacket_connectData_initializer */
    MQTTPacket_connectData condata = MQTTPacket_connectData_initializer;

    rt_memset(&client, 0, sizeof(MQTTClient));

    /* config MQTT context param */
    {
        client.uri = MQTT_TEST_SERVER_URI;

        /* config connect param */
        memcpy(&client.condata, &condata, sizeof(condata));
        client.condata.clientID.cstring = MQTT_CLIENTID;
        client.condata.keepAliveInterval = 60;
        client.condata.cleansession = 1;
        //client.condata.username.cstring = MQTT_USERNAME;
        //client.condata.password.cstring = MQTT_PASSWORD;

        /* config MQTT will param. */
        client.condata.willFlag = 1;
        client.condata.will.qos = MQTT_TEST_QOS;
        client.condata.will.retained = 0;
        client.condata.will.topicName.cstring = MQTT_PUBTOPIC;
        client.condata.will.message.cstring = MQTT_WILLMSG;

        /* malloc buffer. */
        client.buf_size = client.readbuf_size = MQTT_PUB_SUB_BUF_SIZE;
        client.buf = rt_malloc(client.buf_size);
        client.readbuf = rt_malloc(client.readbuf_size);
        if (!(client.buf && client.readbuf))
        {
            rt_kprintf("no memory for MQTT client buffer!\n");
            return;
        }

        /* set event callback function */
        //client.connect_callback = mqtt_connect_callback;
        //client.online_callback = mqtt_online_callback;
        //client.offline_callback = mqtt_offline_callback;

        /* set subscribe table and event callback */
        client.messageHandlers[0].topicFilter = rt_strdup(MQTT_SUBTOPIC);
        client.messageHandlers[0].callback = mqtt_sub_callback;
        client.messageHandlers[0].qos = MQTT_TEST_QOS;

        /* set default subscribe event callback */
        client.defaultMessageHandler = mqtt_sub_callback;
    }

    /* run mqtt client */
    paho_mqtt_start(&client);
    return;
}

static void thread_pub(void *parameter)
{
    while (1)
    {
        /* 阻塞mqtt信号量，等到信号量后发布数据 */
        rt_sem_take(&mqtt_sem, RT_WAITING_FOREVER);

        char pub_data[100] = {0};
        //char *loc_topic;
        rt_strncpy(pub_data, uart_msg.uart_str, 99);

        /*校验json*/
        cJSON *cjson = cJSON_Parse(pub_data);

        if (!cjson) /*不是json*/
        {
        }
        else /*是json*/
        {
            
            if (!paho_mqtt_publish(&client, QOS1, cJSON_GetObjectItem(cjson, "LOC")->valuestring, pub_data))
            {

                uart_msg.state = 0;
                rt_strncpy(uart_msg.uart_str, "", 99);
            }
            cJSON_Delete(cjson);
        }
    }
}

static void mqtt_test_start(void)
{
    if (test_is_started)
    {
        return;
    }

    mq_start();

    while (!client.isconnected)
    {
        rt_kprintf("Waiting for mqtt connection...\n");
        rt_thread_delay(1000);
    }

    pub_thread_tid = rt_thread_create("pub_thread", thread_pub, RT_NULL, 1024, 8, 100);
    if (pub_thread_tid != RT_NULL)
    {
        rt_thread_startup(pub_thread_tid);
    }

    test_is_started = 1;
    return;
}

static int serial2mqtt(int argc, char *argv[])
{
    rt_err_t ret = RT_EOK;
    char uart_name[RT_NAME_MAX];

    /*初始化进程间结构体*/
    uart_msg.state = 0;
    rt_strncpy(uart_msg.uart_str, "", 99);

    if (argc == 2)
    {
        rt_strncpy(uart_name, DEFAULT_UART_NAME, RT_NAME_MAX);
    }
    else
    {
        rt_strncpy(uart_name, DEFAULT_UART_NAME, RT_NAME_MAX);
    }

    /* 查找系统中的串口设备 */
    serial = rt_device_find(uart_name);
    if (!serial)
    {
        rt_kprintf("find %s failed!\n", uart_name);
        return RT_ERROR;
    }

    /* 初始化信号量 */
    rt_sem_init(&rx_sem, "rx_sem", 0, RT_IPC_FLAG_FIFO);
    rt_sem_init(&mqtt_sem, "mqtt_sem", 0, RT_IPC_FLAG_FIFO);
    /* 以中断接收及轮询发送模式打开串口设备 */
    rt_device_open(serial, RT_DEVICE_FLAG_INT_RX);
    /* 设置接收回调函数 */
    rt_device_set_rx_indicate(serial, uart_input);

    /*启动MQTT部分*/
    mqtt_test_start();

    /* 创建 serial 线程 */
    rt_thread_t thread = rt_thread_create("serial", serial_thread_entry, RT_NULL, 1024, 20, 10);
    /* 创建成功则启动线程 */
    if (thread != RT_NULL)
    {
        rt_thread_startup(thread);
    }
    else
    {
        ret = RT_ERROR;
    }
    return ret;
}

#include <finsh.h>
/* 导出到 msh 命令列表中 */
MSH_CMD_EXPORT(serial2mqtt, mqtt forward uart json);
